package com.dongdongshop.mq;

import com.dongdongshop.service.FreemarkerService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

@Component
@RocketMQMessageListener(topic = "audit_status", consumerGroup = "junhui")
public class FreemarkerConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    @Autowired
    private FreemarkerService freemarkerService;

    @Override
    public void onMessage(String s) {

        System.out.println(s);
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    byte[] body = msg.getBody();

                    try {
                        String s = new String(body, "UTF-8");
                        Long aLong = Long.valueOf(s);
                        freemarkerService.createpage(aLong);
                    } catch (Exception e) {
                        e.printStackTrace();

                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
        }
    }
